package cn.ac.iscas.service.source;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.base.utils.TypeConvertUtil;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static cn.ac.iscas.base.constant.MqConnectorConstant.*;


/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/12/20 14:29
 */
public class RocketMqSource {
    private final static String ID = "rocketmq";
    private final static String CONNECTOR_TYPE_VALUE = "rocketmq";

    private String jobId;
    private String componentId;
    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public RocketMqSource(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public RocketMqSource() {

    }

    public String getSql(Map<String, Object> parameters) {
        String host = ParameterUtils.getRequiredString(parameters, "host");
        String port = String.valueOf(ParameterUtils.getRequiredInt(parameters, "port"));
        String consumerGroup = ParameterUtils.getRequiredString(parameters, "consumerGroup");
        String topic = ParameterUtils.getRequiredString(parameters, "topic");
        String tableWithCatalog = String.format("%s.%s.%s", jobId, jobId, topic);
        List<Map<String, Object>> columns = ParameterUtils.getRequiredList(parameters, "metaColumns");
        this.currentTableName = tableWithCatalog;

        String nameServerAddress = host.replace("http://", "").trim() + ":" + port.trim();
        List<FlinkSourceColumn> flinkColumns = new ArrayList<>();
        StringBuffer withBuf = new StringBuffer("WITH (")
                .append(String.format("'%s' = '%s'", CONNECTOR_TYPE_KEY, CONNECTOR_TYPE_VALUE))
                .append(String.format(",'%s' = '%s'", CONNECTOR_TOPIC_KEY, topic))
                .append(String.format(",'%s' = '%s'", CONNECTOR_CONSUME_GROUP_KEY, consumerGroup))
                .append(String.format(",'%s' = '%s'", CONNECTOR_NAMESRV_ADDR_KEY, nameServerAddress));

        String withStr = withBuf.append(")").toString();
        StringBuffer tableSchemaBuf = new StringBuffer(String.format("CREATE TABLE %s (", tableWithCatalog));

        for (Map<String, Object> column : columns) {
            //tableSchemaBuf.append(String.format("%s %s, ", ParameterUtils.getRequiredString(column, "name"), ParameterUtils.getRequiredString(column, "columnType")));
            tableSchemaBuf.append(String.format("%s %s, ", ParameterUtils.getRequiredString(column, "name"), TypeConvertUtil.mysqlTypeToFlinkTypeString(ParameterUtils.getRequiredString(column, "columnType"))));

            FlinkSourceColumn flinkSourceColumn = new FlinkSourceColumn();
            flinkSourceColumn.setName(ParameterUtils.getRequiredString(column, "name"));
            flinkSourceColumn.setType(TypeConvertUtil.mysqlTypeToFlinkTypeString(ParameterUtils.getRequiredString(column, "columnType")));
            flinkSourceColumn.setPrimaryKey(ParameterUtils.getOrDefaultBoolean(column, "primary", false));
            //flinkSourceColumn.setDefaultVal(column.getDefaultVal());
            flinkColumns.add(flinkSourceColumn);
        }
        String tableSchemaString = tableSchemaBuf.toString();
        if (tableSchemaString.trim().endsWith(",")) {
            tableSchemaString = tableSchemaString.substring(0, tableSchemaString.lastIndexOf(","));
        }
        tableSchemaString += ") ";
        String createTableSql = tableSchemaString + withStr;
        this.currentColumns = flinkColumns;

        updateSource();
        return createTableSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }

}
